package com.bw;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.SplitBatchOp;
import com.alibaba.alink.operator.batch.evaluation.EvalMultiClassBatchOp;
import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.alibaba.alink.operator.batch.utils.UDFBatchOp;
import com.alibaba.alink.operator.common.evaluation.MultiClassMetrics;
import com.alibaba.alink.params.shared.clustering.HasKMeansDistanceType;
import com.alibaba.alink.pipeline.PipelineModel;
import com.alibaba.alink.pipeline.classification.DecisionTreeClassifier;
import com.alibaba.alink.pipeline.classification.KnnClassifier;
import com.alibaba.alink.pipeline.tuning.*;
import org.apache.flink.table.functions.ScalarFunction;

public class Test1 {
    public static void main(String[] args) throws Exception {
        BatchOperator.setParallelism(1);
        // 1.使用Flink,创建流式/批式环境，加载指定数据提取出特征值和目标值，把label转换为离散值（5分）
        String filePath = "datafile/lpsa.data";
        String schema
                = "label double,f0 double,f1 double,f2 double,f3 double,f4 double,f5 double," +
                "f6 double,f7 double";
        CsvSourceBatchOp csvSource = new CsvSourceBatchOp()
                .setFilePath(filePath)
                .setSchemaStr(schema)
                .setFieldDelimiter(",");
//        csvSource.print();


        String[] features = {"f0","f1","f2","f3","f4","f5","f6","f7"};
        String label="label";

        // 编码
        UDFBatchOp udfBatchOp = new UDFBatchOp()
                .setFunc(new inOverloadedFunction())
                .setSelectedCols("label")
                .setOutputCol("newLabel")
                .linkFrom(csvSource);

//        udfBatchOp.print();

        // 使用Alink API 把数据划分为，训练集，验证集，测试集（5分）
        // 6  2 2
        BatchOperator<?> spliter = new SplitBatchOp().setFraction(0.6);
        BatchOperator<?> train_data = spliter.linkFrom(udfBatchOp);
        BatchOperator<?> middle_data = spliter.getSideOutput(0);


        BatchOperator<?> spliter1 = new SplitBatchOp().setFraction(0.5);
        BatchOperator<?> test_data = spliter1.linkFrom(middle_data);
        BatchOperator<?> validate_data = spliter1.getSideOutput(0);

        System.out.println("train_data.count() = " + train_data.count());
        System.out.println("test_data.count() = " + test_data.count());
        System.out.println("validate_data.count() = " + validate_data.count());


        //创建模型，（从逻辑回归算法、K近邻算法、Adaboost算法、决策树算法及随机森林）任选两种算法，使用训练数据trainData，分别构建不同分类模型，合理设置参数值;（5分）

        //knn
        KnnClassifier knn = new KnnClassifier()
                .setFeatureCols(features)
                .setPredictionCol("pred")
                .setPredictionDetailCol("pred_detail")
                .setLabelCol(label)
                .setK(3);

        BatchOperator<?> result = knn.fit(train_data).transform(test_data);
        result.print();

        //决策树 不能用向量
        DecisionTreeClassifier tree = new DecisionTreeClassifier()
                .setNumThreads(5)
                .setPredictionDetailCol("pred_detail")
                .setPredictionCol("pred")
                .setLabelCol(label)
                .setFeatureCols(features);
        BatchOperator<?> result2 = tree.fit(train_data).transform(test_data);
//        result2.print();

        //4.使用验证集对上面的模型进行预测评估不同的模型性能，评估准确率和召回率（5分）
        EvalMultiClassBatchOp evop1 = new EvalMultiClassBatchOp().setLabelCol("label").setPredictionDetailCol(
                "pred_detail");

        // 对KNN评估
        MultiClassMetrics lr_metrics = evop1.linkFrom(result).collectMetrics();
        System.out.println("Prefix0 accuracy:" + lr_metrics.getAccuracy());
        System.out.println("Micro Recall:" + lr_metrics.getMicroRecall());

        // 对决策树评估
        MultiClassMetrics lr_metrics1 = evop1.linkFrom(result2).collectMetrics();
        System.out.println("Prefix0 accuracy:" + lr_metrics1.getAccuracy());
        System.out.println("Micro Recall:" + lr_metrics1.getMicroRecall());



//        BinaryClassificationTuningEvaluator tuningEval1 = new BinaryClassificationTuningEvaluator() //用于在网格搜索中评估逻辑回归和决策树
//                .setLabelCol("label")
//                .setPredictionDetailCol("pred_detail")
//                .setTuningBinaryClassMetric("ACCURACY");


//        5.对以上两种算法进行参数调优，训练预测评估模型，获取最佳模型（5分）
        MultiClassClassificationTuningEvaluator tuningEval2 = new MultiClassClassificationTuningEvaluator()//用于在网格搜索中评估knn
                .setLabelCol(label)
                .setPredictionDetailCol("pred_detail")
                .setTuningMultiClassMetric("ACCURACY");


        // 模型优化-KNN
        ParamGrid knn_paramGrid = new ParamGrid()
                .addGrid(knn, KnnClassifier.K, new Integer[] {2,3})
                .addGrid(knn, KnnClassifier.DISTANCE_TYPE, new HasKMeansDistanceType.DistanceType[] {HasKMeansDistanceType.DistanceType.EUCLIDEAN, HasKMeansDistanceType.DistanceType.COSINE});
        GridSearchCV knn_cv = new GridSearchCV()
                .setEstimator(knn)
                .setParamGrid(knn_paramGrid)
                .setTuningEvaluator(tuningEval2)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");
        GridSearchCVModel knn_model = knn_cv.fit(train_data);
        PipelineModel knn_best_model = knn_model.getBestPipelineModel();


        //决策树
        ParamGrid tree_paramGrid = new ParamGrid()
                .addGrid(tree, DecisionTreeClassifier.MIN_SAMPLES_PER_LEAF, new Integer[] {1,3})
                .addGrid(tree, DecisionTreeClassifier.MIN_INFO_GAIN, new Double[] {0.0,0.2});
        GridSearchCV tree_cv = new GridSearchCV()
                .setEstimator(tree)
                .setParamGrid(tree_paramGrid)
                .setTuningEvaluator(tuningEval2)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");
        GridSearchCVModel tree_model = tree_cv.fit(train_data);
        PipelineModel tree_best_model = tree_model.getBestPipelineModel();


        BatchOperator<?> knnBestResult = knn_best_model.transform(test_data);
        BatchOperator<?> treeBestResult = tree_best_model.transform(test_data);


        // knn
        MultiClassMetrics metrics3 = new EvalMultiClassBatchOp()
                .setLabelCol(label)
                .setPredictionDetailCol("pred_detail")
                .linkFrom(knnBestResult).collectMetrics();

        // 决策树
        MultiClassMetrics metrics4 = new EvalMultiClassBatchOp()
                .setLabelCol(label)
                .setPredictionDetailCol("pred_detail")
                .linkFrom(treeBestResult).collectMetrics();

        // 判断指标好坏
        //6.使用评估指标较优者，对样本批量预测（5分）
        if(metrics3.getAccuracy()>metrics4.getAccuracy()){
            System.out.println("knn的模型好");
            BatchOperator<?> transform = knn_best_model.transform(validate_data);
            transform.print();
        }else{
            System.out.println("决策树的模型好");
            BatchOperator<?> transform = tree_best_model.transform(validate_data);
            transform.print();
        }
    }

    public static class inOverloadedFunction extends ScalarFunction {
        public Integer eval(double data) {
            if(data>2){
                return 2;
            }else if((data<=2 && data>1)) {
                return 1;
            }
            else {
                return 0;
            }
        }
    }
}
